package felix.rule.service.actors.chain;

import felix.actor.core.TActorCtx;
import felix.actor.core.TActorRef;
import felix.rule.data.msg.*;
import felix.rule.data.plugin.ComponentLifecycleEvent;
import felix.rule.data.plugin.ComponentLifecycleMsg;
import felix.rule.data.plugin.ComponentLifecycleState;
import felix.rule.service.RuleChainService;
import felix.rule.service.actors.ActorSystemContext;
import felix.rule.service.actors.TEntityActorId;
import felix.rule.service.actors.sharde.ComponentMsgProcessor;
import felix.rule.service.impl.DefaultActorService;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.stream.Collectors;


@Slf4j
public class RuleChainActorMessageProcessor extends ComponentMsgProcessor {

    private final TActorRef parent;
    private final TActorRef self;
    private final Map<String, RuleNodeCtx> nodeActors;
    private final Map<String, List<RuleNodeRelation>> nodeRoutes;
    private final RuleChainService service;
    private String ruleChainName;

    private String firstId;
    private RuleNodeCtx firstNode;
    private boolean started;

    RuleChainActorMessageProcessor(String ruleId, RuleChain ruleChain, ActorSystemContext systemContext
            , TActorRef parent, TActorRef self) {
        super(systemContext, ruleId, ruleChain.getId());
        this.ruleChainName = ruleChain.getName();
        this.parent = parent;
        this.self = self;
        this.nodeActors = new HashMap<>();
        this.nodeRoutes = new HashMap<>();
        this.service = systemContext.getRuleChainService();
    }

    @Override
    public String getComponentName() {
        return null;
    }

    @Override
    public void start(TActorCtx context) {
        if (!started) {
            RuleChain ruleChain = service.findRuleChainById(id);
            if (ruleChain != null) {
                List<RuleNode> ruleNodeList = service.getRuleChainNodes(id);
                log.trace("[{}][{}] Starting rule chain with {} nodes", ruleId, id, ruleNodeList.size());
                // Creating and starting the actors;
                for (RuleNode ruleNode : ruleNodeList) {
                    log.trace("[{}][{}] Creating rule node [{}]: {}", id, ruleNode.getId(), ruleNode.getName(), ruleNode);
                    TActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
                    nodeActors.put(ruleNode.getId(), new RuleNodeCtx(ruleId, self, ruleNodeActor, ruleNode));
                }
                initRoutes(ruleId, ruleChain, ruleNodeList);
                started = true;
            }
        } else {
            onUpdate(context);
        }
    }

    @Override
    public void onUpdate(TActorCtx context) {
        RuleChain ruleChain = service.findRuleChainById(id);
        if (ruleChain != null) {
            ruleChainName = ruleChain.getName();
            List<RuleNode> ruleNodeList = service.getRuleChainNodes(id);
            log.trace("[{}][{}] Updating rule chain with {} nodes", ruleId, id, ruleNodeList.size());
            for (RuleNode ruleNode : ruleNodeList) {
                RuleNodeCtx existing = nodeActors.get(ruleNode.getId());
                if (existing == null) {
                    log.trace("[{}][{}] Creating rule node [{}]: {}", id, ruleNode.getId(), ruleNode.getName(), ruleNode);
                    TActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
                    nodeActors.put(ruleNode.getId(), new RuleNodeCtx(ruleId, self, ruleNodeActor, ruleNode));
                } else {
                    log.trace("[{}][{}] Updating rule node [{}]: {}", id, ruleNode.getId(), ruleNode.getName(), ruleNode);
                    existing.setSelf(ruleNode);
                    existing.getSelfActor().tellWithHighPriority(new ComponentLifecycleMsg(ruleId, RuleNode.class.getSimpleName(),
                            existing.getSelf().getId(), ComponentLifecycleEvent.UPDATED));
                }
            }

            Set<String> existingNodes = ruleNodeList.stream().map(RuleNode::getId).collect(Collectors.toSet());
            List<String> removedRules = nodeActors.keySet().stream().filter(node -> !existingNodes.contains(node)).collect(Collectors.toList());
            removedRules.forEach(ruleNodeId -> {
                log.trace("[{}][{}] Removing rule node [{}]", ruleId, id, ruleNodeId);
                RuleNodeCtx removed = nodeActors.remove(ruleNodeId);
                removed.getSelfActor().tellWithHighPriority(new ComponentLifecycleMsg(ruleId, RuleNode.class.getSimpleName(),
                        removed.getSelf().getId(), ComponentLifecycleEvent.DELETED));
            });

            initRoutes(ruleId, ruleChain, ruleNodeList);
        }
    }

    @Override
    public void stop(TActorCtx ctx) {
        log.trace("[{}][{}] Stopping rule chain with {} nodes", ruleId, id, nodeActors.size());
        nodeActors.values().stream().map(RuleNodeCtx::getSelfActor).map(TActorRef::getActorId).forEach(ctx::stop);
        nodeActors.clear();
        nodeRoutes.clear();
        started = false;
    }

    private TActorRef createRuleNodeActor(TActorCtx ctx, RuleNode ruleNode) {
        return ctx.getOrCreateChildActor(new TEntityActorId(ruleNode.getId(), RuleNodeActor.class.getSimpleName()),
                () -> DefaultActorService.RULE_DISPATCHER_NAME,
                () -> new RuleNodeActor.ActorCreator(systemContext, ruleId, id, ruleChainName, ruleNode.getId()));
    }

    private void initRoutes(String ruleId, RuleChain ruleChain, List<RuleNode> ruleNodeList) {
        nodeRoutes.clear();
        // Populating the routes map;
        for (RuleNode ruleNode : ruleNodeList) {
            List<EntityRelation> relations = service.getRuleNodeRelations(ruleNode.getId());
            log.trace("[{}][{}][{}] Processing rule node relations [{}]", this.ruleId, id, ruleNode.getId(), relations.size());
            if (relations.size() == 0) {
                nodeRoutes.put(ruleNode.getId(), Collections.emptyList());
            } else {
                for (EntityRelation relation : relations) {
                    log.trace("[{}][{}][{}] Processing rule node relation [{}]", this.ruleId, id, ruleNode.getId(), relation.getToId());
                    if (EntityRelation.NODE_TYPE.equals(relation.getEntityType())) {
                        RuleNodeCtx ruleNodeCtx = nodeActors.get(relation.getToId());
                        if (ruleNodeCtx == null) {
                            throw new IllegalArgumentException("Rule Node [" + relation.getFromId() + "] has invalid relation to Rule node [" + relation.getToId() + "]");
                        }
                    }
                    nodeRoutes.computeIfAbsent(ruleNode.getId(), k -> new ArrayList<>())
                            .add(new RuleNodeRelation(ruleNode.getId(), relation.getToId(), relation.getEntityType(), relation.getType()));
                }
            }
        }

        firstId = ruleChain.getFirstRuleNodeId();
        firstNode = nodeActors.get(firstId);
        state = ComponentLifecycleState.ACTIVE;
    }

    void onQueueToRuleEngineMsg(QueueToRuleEngineMsg envelope) {
        EngineMsg msg = envelope.getEngineMsg();
        log.trace("[{}][{}] Processing message [{}]: {}", id, firstId, msg.getId(), msg);
        if (envelope.getRelationTypes() == null || envelope.getRelationTypes().isEmpty()) {
            try {
                checkActive(envelope.getEngineMsg());
                String targetId = msg.getRuleNodeId();
                RuleNodeCtx targetCtx;
                if (targetId == null) {
                    targetCtx = firstNode;
                    msg = msg.copyWithRuleChainId(id);
                } else {
                    targetCtx = nodeActors.get(targetId);
                }
                if (targetCtx != null) {
                    log.trace("[{}][{}] Pushing message to target rule node", id, targetId);
                    pushMsgToNode(targetCtx, msg, "");
                } else {
                    log.trace("[{}][{}] Rule node does not exist. Probably old message", id, targetId);
                    msg.getCallback().onSuccess();
                }
            } catch (RuleNodeException rne) {
                envelope.getEngineMsg().getCallback().onFailure(rne);
            } catch (Exception e) {
                envelope.getEngineMsg().getCallback().onFailure(new RuleEngineException(e.getMessage()));
            }
        } else {
            onTellNext(envelope.getEngineMsg(), envelope.getEngineMsg().getRuleNodeId(), envelope.getRelationTypes(), envelope.getFailureMessage());
        }
    }

    void onRuleChainToRuleChainMsg(RuleChainToRuleChainMsg envelope) {
        try {
            checkActive(envelope.getMsg());
            if (firstNode != null) {
                pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
            } else {
                envelope.getMsg().getCallback().onSuccess();
            }
        } catch (RuleNodeException e) {
            log.debug("Rule Chain is not active. processor [{}] tenant [{}]", id, ruleId);
        }
    }

    void onTellNext(RuleNodeToRuleChainTellNextMsg envelope) {
        onTellNext(envelope.getMsg(), envelope.getOriginator(), envelope.getRelationTypes(), envelope.getFailureMessage());
    }

    private void onTellNext(EngineMsg msg, String originatorNodeId, Set<String> relationTypes, String failureMessage) {
        try {
            checkActive(msg);
            String entityId = msg.getOriginator();
//            TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, msg.getQueueName(), ruleId, entityId);
            List<RuleNodeRelation> relations = nodeRoutes.get(originatorNodeId).stream()
                    .filter(r -> contains(relationTypes, r.getType()))
                    .collect(Collectors.toList());
            int relationsCount = relations.size();
            if (relationsCount == 0) {
                log.trace("[{}][{}][{}] No outbound relations to process", ruleId, entityId, msg.getId());
                if (relationTypes.contains("FAILURE")) {
                    RuleNodeCtx ruleNodeCtx = nodeActors.get(originatorNodeId);
                    if (ruleNodeCtx != null) {
                        msg.getCallback().onFailure(new RuleNodeException(failureMessage, ruleChainName, ruleNodeCtx.getSelf()));
                    } else {
                        log.debug("[{}] Failure during message processing by Rule Node [{}]. Enable and see debug events for more info", entityId, originatorNodeId);
                        msg.getCallback().onFailure(new RuleEngineException("Failure during message processing by Rule Node [" + originatorNodeId + "]"));
                    }
                } else {
                    // 流程结束?
                    msg.getCallback().onSuccess();
                }
            } else if (relationsCount == 1) {
                for (RuleNodeRelation relation : relations) {
                    log.trace("[{}][{}][{}] Pushing message to single target: [{}]", ruleId, entityId, msg.getId(), relation.getOut());
                    pushToTarget(msg, relation.getOut(), relation.getEntityType(), relation.getType());
                }
            } else {
                log.trace("[{}][{}][{}] Pushing message to multiple targets: [{}]", ruleId, entityId, msg.getId(), relations);
                for (RuleNodeRelation relation : relations) {
                    pushToTarget(msg, relation.getOut(), relation.getEntityType(), relation.getType());
                }
            }
        } catch (RuleNodeException rne) {
            msg.getCallback().onFailure(rne);
        } catch (Exception e) {
            msg.getCallback().onFailure(new RuleEngineException("onTellNext - " + e.getMessage()));
        }
    }

    /**
     * 取消 TopicPartitionInfo
     */
    private void pushToTarget(EngineMsg msg, String target, String entityType, String fromRelationType) {
        switch (entityType) {
            case EntityRelation.NODE_TYPE:
                pushMsgToNode(nodeActors.get(target), msg, fromRelationType);
                break;
            case EntityRelation.CHAIN_TYPE:
                parent.tell(new RuleChainToRuleChainMsg(target, id, msg, fromRelationType));
                break;
            default:
                break;
        }
    }

    private boolean contains(Set<String> relationTypes, String type) {
        if (relationTypes == null) {
            return true;
        }
        for (String relationType : relationTypes) {
            if (relationType.equalsIgnoreCase(type)) {
                return true;
            }
        }
        return false;
    }

    private void pushMsgToNode(RuleNodeCtx nodeCtx, EngineMsg msg, String fromRelationType) {
        if (nodeCtx != null) {
            nodeCtx.getSelfActor().tell(new RuleChainToRuleNodeMsg(new DefaultTContext(systemContext, nodeCtx), msg, fromRelationType));
        } else {
            log.error("[{}][{}] RuleNodeCtx is empty", id, ruleChainName);
            msg.getCallback().onFailure(new RuleEngineException("Rule Node CTX is empty"));
        }
    }

    @Override
    protected RuleNodeException getInactiveException() {
        RuleNode firstRuleNode = firstNode != null ? firstNode.getSelf() : null;
        return new RuleNodeException("Rule Chain is not active!  Failed to initialize.", ruleChainName, firstRuleNode);
    }
}
